package cn.doitedu.demo11;

import cn.doitedu.pojo.UserEvent;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/20
 * @Desc: 学大数据，上多易教育
 *   本规则模型，支持如下规则：
 *      规则a: 用户发生X行为，就命中规则
 *      规则b: 用户发生Q行为，就命中规则
 *      规则c: 用户发生S行为，就命中规则
 *
 **/
public class RuleModel_1_Calculator implements RuleCalculator {

    String paramEventId;
    String ruleId;
    JSONObject message = new JSONObject();

    // {"rule_id":"rule-1-001","event_id":"G"}
    @Override
    public void init(RuntimeContext runtimeContext,
                     String ruleParamJson,
                     Roaring64Bitmap preSelectUsersBitmap,
                     Long webCalcEndTime) throws IOException {
        JSONObject jsonObject = JSONObject.parseObject(ruleParamJson);
        paramEventId = jsonObject.getString("event_id");
        ruleId = jsonObject.getString("rule_id");

    }

    @Override
    public void calc(UserEvent eventBean, Collector<String> out) throws IOException {

        if(eventBean.getEvent_id().equals(paramEventId)){

            message.put("user_id",eventBean.getUid());
            message.put("hit_time",eventBean.getTimestamp());
            message.put("rule_id",ruleId);

            out.collect(message.toJSONString());
        }



    }
}
